package edu.ustb.efws.pipeline.collect;

import com.stoyanr.evictor.ConcurrentMapWithTimedEviction;
import com.stoyanr.evictor.map.ConcurrentHashMapWithTimedEviction;
import edu.ustb.efws.base.domain.Detector;
import edu.ustb.efws.pipeline.collect.handler.CollectChina;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 采集工作者
 *
 * @author TS Group
 */
public class CollectWorker {
    private static final Logger LOGGER = LoggerFactory.getLogger(CollectWorker.class);
    private static final AtomicInteger WORKER_COUNTER = new AtomicInteger(0);

    private final String id;
    private final ExecutorService executorService;
    private final CollectChina collectChina;
    private final CollectService collectService;
    private final BlockingQueue<Detector> queue;
    private final ConcurrentMapWithTimedEviction<Integer, Integer> errorCollects;

    /**
     * 构造{@link CollectWorker}
     *
     * @param collectChina {@link CollectChina}
     * @param collectService {@link CollectService}
     * @param threadFactory {@link ThreadFactory}
     */
    public CollectWorker(CollectChina collectChina,
                         CollectService collectService,
                         ThreadFactory threadFactory){

        this.id = buildId();
        this.collectChina = collectChina;
        this.collectService = collectService;
        this.executorService = Executors.newSingleThreadExecutor(threadFactory);
        this.queue = new LinkedBlockingQueue<>(200);
        errorCollects = new ConcurrentHashMapWithTimedEviction<>();
        //自动开始
        start();
    }

    /**
     * 生成分发工作编号
     *
     * @return 分发工作编号
     */
    private String buildId(){
        return String.format("collect-worker-%d", WORKER_COUNTER.incrementAndGet());
    }

    /**
     * 开始分发任务
     */
    private void start(){
        executorService.submit(() -> {
            for(;;){
                try{
                    collectHandle(queue.take());
                }catch (InterruptedException e) {
                    LOGGER.error("Dispatcher interrupted throw={}", e.getMessage());
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        });
    }

    private void collectHandle(Detector o){

        if(errorCollects.containsKey(o.getId())){
            return ;
        }

        try{
            collectService.collect(o).ifPresent(e ->{
                long startMills = System.currentTimeMillis();
                collectChina.doHandle(e);
                long useTime = System.currentTimeMillis()-startMills;
                LOGGER.trace("Start collection detector workId={}, id={}, name={}, useMills={}", id, o.getId(), o.getDescription(), useTime);
            });
        }catch (Exception e){
            LOGGER.error("Collect handle fail workId={} detector={}, error={}", id, o.toString(), e.getMessage());
            //暂停5分钟再采集
            errorCollects.put(o.getId(), o.getId(), Duration.ofMinutes(5).toMillis());
        }
    }

    /**
     * 添加分发消息到队列中
     *
     * @param o 分发对象
     */
    public boolean push(Detector o)throws InterruptedException{
        if(errorCollects.containsKey(o.getId())){
            return true;
        }
        return queue.offer(o, 200, TimeUnit.MILLISECONDS);
    }

    /**
     * 分发工作编号
     *
     * @return 编号
     */
    public String getId() {
        return id;
    }

    /**
     * 关闭分发任务
     */
    void shutdown(){
        if(executorService.isShutdown()){
            executorService.shutdown();
            LOGGER.info("Collect work shutdown id={}", id);
        }
    }
}
